package DataCollect;

import org.apache.commons.io.FileUtils;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.EventDrivenSource;
import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.conf.Configurable;
import org.apache.flume.event.EventBuilder;
import org.apache.flume.source.AbstractSource;

import java.io.*;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

public class MySource extends AbstractSource implements Configurable, EventDrivenSource {
    //定义成员属性
    //原始数据文件
    private String filePath;
    //偏移量文件
    private String posFile;
    //编码格式
    private String charset;
    //睡眠间隔时间
    private Long sleepTime = 0L;
    //创建线程的对象
    private FileRunnable fileRunnable;

    //private ExecutorService executor;

    /*
    * 获取配置文件的参数
    * */
    @Override
    public void configure(Context context) {
        filePath = context.getString("filePath");
        posFile = context.getString("posFile");
        charset = context.getString("charset","utf-8");
        sleepTime = context.getLong("sleepTime");
        //charset = Charset.forName(context.getString("charset"));
        //bufferCount = context.getInteger("bufferCount");

    }
    //执行主要的业务逻辑
    @Override
    public synchronized void start() {
        //单线程线程池
         ExecutorService executor = Executors.newSingleThreadExecutor();

        //发送event
        ChannelProcessor channelProcessor = getChannelProcessor();
        //创建runnable接口实现类对象
        fileRunnable = new FileRunnable(filePath,posFile,charset,sleepTime,channelProcessor);


        executor.submit(fileRunnable);
        //executor.execute(fileRunnable);
        //线程池提交任务
         super.start();
    }

    @Override
    public synchronized void stop() {
        /*if (executor!=null){
            try {
                fileRunnable.setFlag(false);
                Thread.sleep(sleepTime);
                executor.shutdown();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }*/
        super.stop();
    }

    public class FileRunnable implements Runnable{
        //原始数据文件
        private String filePath;
        //偏移量文件
        private String posFile;
        //编码格式
        private String charset;
        //偏移量
        private Long offset = 0L;
        private Long interval;
        //原始文件随机读取类
        private RandomAccessFile randomAccessFile;

        private boolean flag = true;
        private File file;
        private ChannelProcessor channelProcessor;
        //private Integer bufferCount = 5;

        public void setFlag(boolean flag) {
            this.flag = flag;
        }


        public FileRunnable(String filePath,String posFile,String charset,Long interval,ChannelProcessor channelProcessor){
            this.filePath = filePath;
            this.posFile = posFile;
            this.charset = charset;
            this.interval = interval;
            this.channelProcessor = channelProcessor;

            file = new File(posFile);
            if (!file.exists()){
                try {
                    file.createNewFile();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }

            try {
                String offsetStr = FileUtils.readFileToString(file);
                if (offsetStr != null && !"".equals(offsetStr.trim())){
                    offset = Long.parseLong(offsetStr);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }

            try {
                randomAccessFile = new RandomAccessFile(filePath, "r");
                randomAccessFile.seek(offset);

            } catch (IOException e) {
                e.printStackTrace();
            }

        }

        @Override
        public void run() {
            //final List<Event> eventList = new ArrayList<Event>();
            while (flag){
                try {
                    String line = randomAccessFile.readLine();
                    if (line!=null){
                        //eventList.add(EventBuilder.withBody(line.getBytes(charset)));


                        //if (eventList.size()>=bufferCount){
                        channelProcessor.processEvent(EventBuilder.withBody(line.getBytes(charset)));
                        offset = randomAccessFile.getFilePointer();
                        FileUtils.writeStringToFile(file, offset.toString());

                    }else {
                        Thread.sleep(interval);
                    }
                } catch (IOException  e) {
                    e.printStackTrace();
                } catch (InterruptedException e){
                    e.printStackTrace();
                }
            }
        }
    }
}
